package nsqio

import (
	"encoding/json"
	"errors"
	"gitee.com/liukexing/mqbroker/core/driver/logger"
	"gitee.com/liukexing/mqbroker/core/typedef"
	"github.com/nsqio/go-nsq"
	"strconv"
	"sync"
	"time"
)

type NsqPubSub struct {
	typedef.PubSubServer
	Publisher  sync.Map
	Subscriber sync.Map
	ReqMap     sync.Map
	ReplyFunc  sync.Map
}

func (h *NsqPubSub) GetPubConfig() interface{} {
	for _, v := range pubConfig {
		return v
	}
	return nil
}

func (h *NsqPubSub) GetSubConfig() interface{} {
	for _, v := range subConfig {
		return v
	}
	return nil
}

func (N *NsqPubSub) NewPublisher(name string, config interface{}, topic string) bool {
	if _, ok := N.Publisher.Load(name); ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Publisher has exist"),
		})
		return true
	}
	nsqconfig := config.(typedef.NsqConfig)
	producer, err := nsq.NewProducer(nsqconfig.HostName+":"+strconv.Itoa(nsqconfig.Tcp), nsq.NewConfig())
	if logger.CheckErr(typedef.ErrInfo{
		Info: Config.Info,
		Err:  err,
	}) {
		return true
	}
	producer.SetLogger(nil, 0)
	producer.Ping()
	N.Publisher.Store(name, &typedef.NsqPub{
		NsqProducer: producer,
		Topic:       topic,
	})
	return false
}

func (N *NsqPubSub) Publish(name string, message string) bool {
	if v, ok := N.Publisher.Load(name); !ok || v == nil {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Publisher Not Found"),
		})
		return true
	} else {
		nsqpub := v.(*typedef.NsqPub)
		if nsqpub.NsqProducer != nil {
			if message == "" {
				logger.Warn(typedef.CommInfo{
					Info:    Config.Info,
					Message: "Message is Empty",
				})
				return true
			}
			return logger.CheckErr(typedef.ErrInfo{
				Info: Config.Info,
				Err:  nsqpub.NsqProducer.Publish(nsqpub.Topic, []byte(message)),
			})
		}
	}

	return false
}

func (N *NsqPubSub) DeferredPublish(name string, message string, delay time.Duration) bool {
	if v, ok := N.Publisher.Load(name); !ok || v == nil {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Publisher Not Found"),
		})
		return true
	} else {
		nsqpub := v.(*typedef.NsqPub)
		if nsqpub.NsqProducer != nil {
			if message == "" {
				logger.Warn(typedef.CommInfo{
					Info:    Config.Info,
					Message: "Message is Empty",
				})
				return true
			}
			return logger.CheckErr(typedef.ErrInfo{
				Info: Config.Info,
				Err:  nsqpub.NsqProducer.DeferredPublish(nsqpub.Topic, delay, []byte(message)),
			})
		}
	}
	return false
}

func (N *NsqPubSub) DelPublisher(name string) bool {
	if v, ok := N.Publisher.Load(name); !ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Publisher Not Found"),
		})
		return true
	} else {
		nsqpub := v.(*typedef.NsqPub)
		if nsqpub.NsqProducer != nil {
			nsqpub.NsqProducer.Stop()
			N.Publisher.Delete(name)
		}
	}

	return false
}

func (N *NsqPubSub) NewSubscriber(name string, config interface{}, topic string) bool {
	if _, ok := N.Subscriber.Load(name); ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Subscriber has exist"),
		})
		return true
	}
	nsqconfig := config.(typedef.NsqConfig)
	N.Subscriber.Store(name, &typedef.NsqSub{
		Topic: topic,
		Addr:  nsqconfig.HostName + ":" + strconv.Itoa(nsqconfig.Http),
	})
	return false
}

func (N *NsqPubSub) AddSubscriber(channel string, name string, handel typedef.MessageHandle) bool {
	if v, ok := N.Subscriber.Load(name); !ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Subscriber Not Found"),
		})
		return true
	} else {
		nsqsub := v.(*typedef.NsqSub)
		cfg := nsq.NewConfig()
		cfg.LookupdPollInterval = time.Second
		cfg.MaxInFlight = len(pubConfig) + 1
		c, err := nsq.NewConsumer(nsqsub.Topic, channel, cfg) // 新建一个消费者
		if logger.CheckErr(typedef.ErrInfo{
			Info: Config.Info,
			Err:  err,
		}) {
			return true
		}
		c.SetLogger(nil, 0)

		c.AddHandler(&NsqHandle{handel: handel}) // 添加消费者接口
		if logger.CheckErr(typedef.ErrInfo{
			Info: Config.Info,
			Err:  c.ConnectToNSQLookupd(nsqsub.Addr),
		}) {
			return true
		}
		nsqsub.NsqConsumer.Store(channel, c)
	}

	return false
}
func (N *NsqPubSub) RemoveSubscriber(channel string, name string) bool {
	if v, ok := N.Subscriber.Load(name); !ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Subscriber Not Found"),
		})
		return true
	} else {
		nsqsub := v.(*typedef.NsqSub)
		if c, ok := nsqsub.NsqConsumer.Load(channel); !ok {
			logger.Error(typedef.ErrInfo{
				Info: Config.Info,
				Err:  errors.New("Channel Not Found"),
			})
			return true
		} else {
			Channel := c.(*nsq.Consumer)
			if Channel != nil {
				Channel.Stop()
				nsqsub.NsqConsumer.Delete(channel)
			}
		}

	}
	return false
}
func (N *NsqPubSub) DelSubscriber(name string) bool {
	if v, ok := N.Subscriber.Load(name); !ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Subscriber Not Found"),
		})
		return true
	} else {
		nsqsub := v.(*typedef.NsqSub)
		cnt := 0
		nsqsub.NsqConsumer.Range(func(key, value interface{}) bool {
			cnt++
			return false
		})
		if cnt > 0 {
			logger.Error(typedef.ErrInfo{
				Info: Config.Info,
				Err:  errors.New("NsqConsumer Not All Removed"),
			})
			return true
		}
		N.Subscriber.Delete(name)
	}

	return false
}

func (N *NsqPubSub) Request(name string, message interface{}, timeout time.Duration) typedef.ResponseData {
	if N.checkReply() {
		return typedef.ResponseData{
			Error:   true,
			Message: "checkReply Failed",
		}
	}

	reqid := typedef.GetUUID()
	data := typedef.RequestData{
		Topic: name,
		Data:  message,
		ReqID: reqid,
	}
	jsv, err := json.Marshal(data)
	if logger.CheckErr(typedef.ErrInfo{
		Info: Config.Info,
		Err:  err,
	}) {
		return typedef.ResponseData{
			Error:   true,
			Message: "Json Marshal Failed",
		}
	}
	if N.Publish("Request", string(jsv)) {
		return typedef.ResponseData{
			Error:   true,
			Message: "Publish Message Failed",
		}
	}
	rsp := make(chan typedef.ReplyData)
	key := name + reqid
	//N.ReqMap[key] = rsp
	N.ReqMap.Store(key, rsp)
	select {
	case response := <-rsp:
		//delete(N.ReqMap, key)
		N.ReqMap.Delete(key)
		if response.Error {
			return typedef.ResponseData{
				Error:   true,
				Message: response.Data.(string),
			}
		} else {
			return typedef.ResponseData{
				Error: false,
				Data:  response.Data,
			}
		}
	case <-time.After(timeout):
		//delete(N.ReqMap, key)
		N.ReqMap.Delete(key)
		return typedef.ResponseData{
			Error:   true,
			Message: "Request Message Timeout",
		}

	}
	//delete(N.ReqMap, key)
	N.ReqMap.Delete(key)
	return typedef.ResponseData{
		Error:   true,
		Message: "Error Back",
	}
}
func (N *NsqPubSub) AddReply(name string, handel typedef.ReplyHandle) bool {
	if N.checkReply() {
		return true
	}

	if _, ok := N.ReplyFunc.Load(name); ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("ReplyFunc Has Exist"),
		})
		return true
	}
	N.ReplyFunc.Store(name, handel)
	return false
}
func (N *NsqPubSub) RemoveReply(name string) bool {
	if _, ok := N.ReplyFunc.Load(name); !ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("ReplyFunc Not Found"),
		})
		return true
	}
	N.ReplyFunc.Delete(name)
	return false
}

func (N *NsqPubSub) checkReply() bool {
	reqid := typedef.GetUUID()
	subconfig := N.GetSubConfig()
	if _, ok := N.Publisher.Load("Request"); !ok {
		if N.NewPublisher("Request", N.GetPubConfig(), "Request") {
			return true
		}
	}
	if _, ok := N.Publisher.Load("Response"); !ok {
		if N.NewPublisher("Response", N.GetPubConfig(), "Response") {
			return true
		}
	}
	if _, ok := N.Subscriber.Load("Request"); !ok {
		if N.NewSubscriber("Request", subconfig, "Request") {
			return true
		}
		if N.AddSubscriber("Handle_"+reqid, "Request", &NsqRequestHandle{parent: N}) {
			return true
		}
	}
	if _, ok := N.Subscriber.Load("Response"); !ok {
		if N.NewSubscriber("Response", subconfig, "Response") {
			return true
		}
		if N.AddSubscriber("Handle_"+reqid, "Response", &NsqResponseHandle{parent: N}) {
			return true
		}
	}
	return false
}
